package cn.xuexiyuan.flinkstudy.connectors;

import cn.xuexiyuan.flinkstudy.entity.Student;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @Description:
 * @Author 左龙龙
 * @Date 21-3-24
 * @Version 1.0
 **/
public class JDBCDemo {

    public static void main(String[] args) throws Exception {
        // 0.env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 1.source
        DataStreamSource<Student> studentDS = env.fromElements(new Student(null, "jdbc sink", 5));

        // 2.transformation

        // 3.sink
        String url = "jdbc:mysql://localhost:3306/flink_study";
        String user = "root";
        String passwd = "";
        studentDS.addSink(JdbcSink.sink(
                "insert into tb_student (name, age) values (?,?)",
                (ps, t) -> {
                    ps.setString(1, t.getName());
                    ps.setInt(2, t.getAge());
                },
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withDriverName("com.mysql.jdbc.Driver")
                        .withUrl(url)
                        .withUsername(user)
                        .withPassword(passwd)
                        .build()));

        // 4.excute
        env.execute("JDBCDemo");    }

}
